package com.xflib.framework.ratelimit.core;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ProducerConsumerExample {
    private static final int MAX_SIZE = 2000;  // 队列最大容量
    private final ConcurrentLinkedDeque<Integer> queue = new ConcurrentLinkedDeque<>();  // 共享队列
    private final Object lock = new Object();  // 显式锁对象[^2]

    // 生产者线程
    class Producer implements Runnable {
        private int produceCount = 0;
        private int discardCount = 0;

        @Override
        public void run() {
            int value = 0;
            while (!Thread.currentThread().isInterrupted()) {
                synchronized (lock) {  // 获取对象锁
                    produceCount++;
                    if (queue.size() == MAX_SIZE) {
                        discardCount++;
                        //System.out.println("队列已满，抛弃产品: " + value + "，生产计数: " + produceCount + "，抛弃计数: " + discardCount);
                    } else {
                        queue.offer(value++);
                        //System.out.println("生产: " + value + "，队列大小: " + queue.size() + "，生产计数: " + produceCount);
                    }
                    lock.notifyAll();  // 唤醒所有等待线程（消费者）
                }
            }
        }
    }

    // 消费者线程
    class Consumer implements Runnable {
        private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private long lastConsumeTime = System.currentTimeMillis();
    private int consumeCount = 0;

    @Override
    public void run() {
        scheduler.scheduleAtFixedRate(() -> {
            synchronized (lock) {
                while (queue.isEmpty()) {
                    try {
                        System.out.println("队列为空，消费者等待");
                        lock.wait();  // 释放锁并进入等待
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        scheduler.shutdown();
                        return;
                    }
                }
                // 每次只消费一个元素（100毫秒间隔）
                int value = queue.poll();
                long currentTime = System.currentTimeMillis();
                long interval = currentTime - lastConsumeTime;
                lastConsumeTime = currentTime;
                consumeCount++;
                //System.out.println("消费: " + value + "，队列大小: " + queue.size() + "，消费间隔: " + interval + "ms，消费计数: " + consumeCount);
                lock.notifyAll();  // 唤醒所有等待线程（生产者）
            }
        }, 0, 100, TimeUnit.MILLISECONDS);
    }

    public void shutdown() {
        scheduler.shutdown();
    }
    }

    public static void main(String[] args) {
        ProducerConsumerExample example = new ProducerConsumerExample();
        long startTime = System.currentTimeMillis();
        Thread producerThread = new Thread(example.new Producer());
        Thread consumerThread = new Thread(example.new Consumer());
        producerThread.start();
        consumerThread.start();
        
        // 等待1分钟后停止线程
        try {
            Thread.sleep(60000);
            producerThread.interrupt();
            consumerThread.interrupt();
            // 等待线程终止
            producerThread.join();
            consumerThread.join();
            System.out.println("程序运行1分钟，已停止");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
